package org.tib.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * An concurrent task execution processor.
 * The implementation uses ExecutoreService to invoke tasks in multiple threads asynchronously.
 * The thread-pool created is a cached pool which would create threads as and when necessary and re-use already
 * created threads.
 *
 * This component provides the API user to provide a ExectorService of their choice. Ideally you want a defined set of
 * threads setup for the async tasks. The user can define an ExecutorService for the life-cycle of the application and
 * provide this component with it. For example if using spring framework you can define an ExecutorService as a bean,
 * and inject that into the component. This would ensure one ExecutorService with a set of threads that the can be reused
 * through out the application life-cycle(at least the life-cycle of the spring bean in this example)       
 *
 * Todo:
 * This is a list of enhancements/improvements that could be incorporated
 * - Allow the API user to specify a time-out for any execution (possible through the Future.get(timeout)
 *
 * @author Imran
 */
public class ConcurrentExecutor<V,R> {
    private List<Processor<V,R>> processors;
    private ExecutorService executorService;

    public List<R> executeAll() throws ConcurrentException {
        List<R> results = new ArrayList<R>();

        if(this.processors == null || this.processors.isEmpty()) {
            throw new ConcurrentException("Cannot run empty processors");
        }
        else {
            try {
                // set invoke all processors through our executor service
                List<Future<R>> futures = getExecutorService().invokeAll(processors);

                // Now gather the results
                for(Future<R> future: futures) {
                    results.add(future.get());
                }
            } catch (InterruptedException iex) {
                throw new ConcurrentException("Execution interrupted", iex);
            } catch (ExecutionException ex) {
                throw new ConcurrentException("Execution terminated with an un-expected exception", ex);
            }
        }
        return results;
    }

    public List<Processor<V,R>> getProcessors() {
        return processors;
    }

    public void setProcessors(List<Processor<V,R>> processors) {
        this.processors = processors;
    }

    public ExecutorService getExecutorService() {
        if(this.executorService != null) {
            return executorService;
        }
        else {
            return Executors.newCachedThreadPool();
        }
    }

    public void setExecutorService(ExecutorService executorService) {
        this.executorService = executorService;
    }
}
